package com.cg;

import com.google.common.collect.Lists;

import java.text.MessageFormat;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 多个completableFuture任务结果集阻塞合并
 */
public class CompleteFutureTest03 {

    private static int i = 0;

    // 执行的cpu核数是4
    private static final int cores = Runtime.getRuntime().availableProcessors();

    private static final int limitSize = 1000000;

    private static final ReentrantLock reentrantLock = new ReentrantLock();

    public static void main(String[] args) {
        long start = System.currentTimeMillis();
        List<CompletableFuture<Void>> objects = Lists.newArrayList();
        for(int k=0;k<limitSize;k++){

            CompletableFuture<Void> voidCompletableFuture = CompletableFuture.runAsync(() -> {
//                System.out.println(Thread.currentThread().getName());
                // 不要将获取锁的过程写在try中，如果在获取锁（自定义锁）时发生异常，
                // 异常抛出同时，也会导致锁无故释放
                reentrantLock.lock();
                try {
                    i += 1;
                }finally {
                    reentrantLock.unlock();
                }
            }, executor);
            objects.add(voidCompletableFuture);
        }
        CompletableFuture.allOf(objects.toArray(new CompletableFuture[limitSize])).join();
        long end = System.currentTimeMillis();
        System.out.println(MessageFormat.format("加锁耗时:{0}ms",(end-start)));
        System.out.println(i);
    }

    private static final ThreadPoolExecutor executor = new ThreadPoolExecutor(cores,
            cores,
            60,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(limitSize),
            new ThreadFactory() {
                final AtomicInteger a = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, "xiaoxu-" + a.getAndAdd(1));
                }
            }, new ThreadPoolExecutor.AbortPolicy());

}
